﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

#region

using System;
using System.Collections.Generic;
using AzureMRCore;
using AzureMRCore.DataTypes;
using AzureMRCore.MapRed;

#endregion

namespace PageRankMR
{
    internal class PageRankMapper :
        Mapper<IntKey, AdjacencyListValue<int>, IntKey, DoubleValue, IntKey, DoubleVectorValue>
    {
        protected override int Map(IntKey key, AdjacencyListValue<int> value, List<KeyValuePair<IntKey, DoubleVectorValue>> dynamicData, IOutputCollector<IntKey, DoubleValue> outputCollector, string programArgs)
        {
            double[] pageRankValues = dynamicData[0].Value.Value;
            //create length+1 to store the pagerank for dangling pages
            double[] outPageRankMap = new double[pageRankValues.Length + 1];
            int numPages = Int32.Parse(programArgs);

            foreach (KeyValuePair<int, List<int>> pageInfo in value.AdjList)
            {
                int pageIndex = pageInfo.Key;
                int outLinkTotal = pageInfo.Value.Count;
                if (outLinkTotal == 0)
                {
                    outPageRankMap[pageRankValues.Length] += pageRankValues[pageIndex];
                    outputCollector.Collect(IntKey.GetInstance(pageRankValues.Length),
                                            DoubleValue.GetInstance(pageRankValues[pageIndex]));
                }
                else
                {
                    foreach (int outLink in pageInfo.Value)
                    {
                        // avoid assigning page ranks to pages outside the sub graph we consider
                        if (outLink < numPages)
                        {
                            DoubleValue newPageRank = DoubleValue.GetInstance(pageRankValues[pageIndex]/outLinkTotal);
                            outputCollector.Collect(IntKey.GetInstance(outLink), newPageRank);
                        }
                    }
                }
            }
            return 0;
        }
    }
}